package org.apache.ocean.server;

import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.apache.commons.lang.StringUtils;
import org.apache.lucene.document.Document;
import org.apache.lucene.index.Term;
import org.apache.lucene.search.Query;
import org.apache.lucene.store.BufferedIndexOutput;
import org.apache.lucene.store.IndexInput;
import org.apache.ocean.CLong;
import org.apache.ocean.ClusterConstants;
import org.apache.ocean.ClusterUtil;
import org.apache.ocean.IDVersion;
import org.apache.ocean.IndexSchema;
import org.apache.ocean.OceanDocument;
import org.apache.ocean.OceanXML;
import org.apache.ocean.SearchIndexShard;
import org.apache.ocean.SearchRequest;
import org.apache.ocean.SearchRequestUtil;
import org.apache.ocean.SearchResponse;
import org.apache.ocean.SearchService;
import org.apache.ocean.Service;
import org.apache.ocean.ShardInfo;
import org.apache.ocean.TermFrequencies;
import org.apache.ocean.TimeoutState;
import org.apache.ocean.TransactionID;
import org.apache.ocean.TransactionRecord;
import org.apache.ocean.UpdateResponse;
import org.apache.ocean.XMLUtil;
import org.apache.ocean.util.PrintStringWriter;
import org.jdom.Element;

/**
 * 
 * @author jasonr
 */
public class SearchServlet extends HttpServlet {
	public static Logger log = Logger.getLogger(SearchServlet.class.getName());
	public SearchService searchService;

	public void doPost(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
		//log.info("SearchServlet get: " + request.getRequestURI());
		String responseXML = null;
		try {
			if (!StringUtils.equals(Service.STARTED, searchService.getServiceStatus())) {
				throw new Exception("search service not started yet serviceStatus: " + searchService.getServiceStatus());
			}
			IndexSchema indexSchema = searchService.getIndexSchema();
			response.setContentType(ClusterConstants.CONTENT_TYPE_XML);
			OceanXML oceanRequest = OceanXML.parse(request.getReader());
			String action = oceanRequest.getHeaderValue("action");
			if (StringUtils.equals("addtransactionrecord", action)) {
				responseXML = handleAddTransactionRecord(oceanRequest);
				return;
			}
			OceanXML oceanResponse = new OceanXML();
			oceanResponse.addHeader("serverInfo", searchService.getServerInfo());
			List<UpdateResponse> updateResponses = new ArrayList<UpdateResponse>();
			for (Element element : XMLUtil.getChildren(oceanRequest.bodyElement)) {
				String name = element.getName();
				String batchID = element.getAttributeValue("batchid");
				TimeoutState timeoutState = new TimeoutState(1000 * 5);
				if (StringUtils.equalsIgnoreCase("update", name)) {
					Element documentElement = XMLUtil.getFirstChild(element);
					OceanDocument oceanDocument = new OceanDocument(documentElement);
					Document document = ClusterUtil.toDocument(oceanDocument, indexSchema);
					UpdateResponse updateResponse = searchService.updateDocumentMaster(document, timeoutState);
					updateResponse.batchID = batchID;
					updateResponses.add(updateResponse);
				} else if (StringUtils.equalsIgnoreCase("add", name)) {
					Element documentElement = XMLUtil.getFirstChild(element);
					OceanDocument oceanDocument = new OceanDocument(documentElement);
					Document document = ClusterUtil.toDocument(oceanDocument, indexSchema);
					UpdateResponse updateResponse = searchService.addDocumentMaster(document, timeoutState);
					updateResponse.batchID = batchID;
					updateResponses.add(updateResponse);
				} else if (StringUtils.equalsIgnoreCase("delete", name)) {
					CLong id = XMLUtil.getChildLong("id", element);
					CLong version = XMLUtil.getChildLong("version", element);
					String shardID = element.getChildText("shardid");
					IDVersion idVersion = new IDVersion(id, version, shardID);
					UpdateResponse updateResponse = searchService.deleteDocumentMaster(idVersion, timeoutState);
					updateResponse.batchID = batchID;
					updateResponses.add(updateResponse);
				}
			}
			for (UpdateResponse updateResponse : updateResponses) {
				oceanResponse.addBody(updateResponse.toElement());
			}
			responseXML = XMLUtil.outputElement(oceanResponse.toElement());
		} catch (Throwable throwable) {
			try {
				responseXML = OceanXML.toOceanXMLString(throwable);
			} catch (Throwable exTh) {
				log.log(Level.SEVERE, "", throwable);
				log.log(Level.SEVERE, "inside creating exception response", exTh);
				throw new ServletException(exTh);
			}
		} finally {
			PrintWriter writer = response.getWriter();
			writer.write(responseXML);
		}
	}
  
	private String handleAddTransactionRecord(OceanXML oceanRequest) throws Exception {
		TransactionRecord transactionRecord = (TransactionRecord)oceanRequest.getFirstInBody(new TransactionRecord.TransactionRecordTransform());
		searchService.saveTransactionRecord(transactionRecord);
		OceanXML oceanResponse = new OceanXML("status", "ok");
		oceanResponse.setServerInfo(searchService.getServerInfo());
		return XMLUtil.outputElement(oceanResponse.toElement());
	}
	
	public void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
		//log.info("SearchServlet get: " + request.getRequestURI());
		PrintStringWriter printStringWriter = new PrintStringWriter();
		try {
			if (!StringUtils.equals(Service.STARTED, searchService.getServiceStatus())) {
				throw new Exception("search service not started yet serviceStatus: " + searchService.getServiceStatus());
			}
			String action = request.getParameter("action");
			if (StringUtils.equals("getindexfile", action)) {
				handleGetIndexFile(request, response);
			} else if (StringUtils.equals("getshardinfos", action)) {
				handleGetShardInfos(request, printStringWriter);
			} else if (StringUtils.equals("gettransactionrecords", action)) {
				handleGetTransactionRecords(request, printStringWriter);
			} else if (StringUtils.equals("gettermfrequencies", action)) {
				handleGetTermFrequencies(request, printStringWriter);
			} else {
				IndexSchema indexSchema = searchService.getIndexSchema();
				response.setContentType(ClusterConstants.CONTENT_TYPE_XML);
				SearchRequestUtil searchRequestUtil = new SearchRequestUtil(request, indexSchema);
				SearchRequest searchRequest = searchRequestUtil.getSearchRequest();
				SearchResponse searchResponse = searchService.search(searchRequest);
				searchResponse.fields = searchRequest.fields;
				OceanXML oceanXML = searchResponse.toOceanXML(indexSchema);
				oceanXML.addHeader("serverInfo", searchService.getServerInfo());
				Element responseElement = oceanXML.toElement();
				XMLUtil.outputElement(responseElement, printStringWriter);
			}
		} catch (Throwable throwable) {
			try {
				log.log(Level.SEVERE, "", throwable);
				String responseXML = OceanXML.toOceanXMLString(throwable);
				printStringWriter = new PrintStringWriter();
				printStringWriter.print(responseXML);
			} catch (Throwable exTh) {
				log.log(Level.SEVERE, "", throwable);
				log.log(Level.SEVERE, "inside creating exception response", exTh);
				throw new ServletException(exTh);
			}
		} finally {
			PrintWriter writer = response.getWriter();
			writer.write(printStringWriter.toString());
		}
	}
	
	private void handleGetIndexFile(HttpServletRequest request, HttpServletResponse response) throws Exception {
		String shardID = request.getParameter("shardid");
		String indexID = request.getParameter("indexid");
		String file = request.getParameter("file");
		String snapshot = request.getParameter("snapshot");
		SearchIndexShard searchIndexShard = searchService.getSearchIndexShard(shardID);
		IndexInput indexInput = searchIndexShard.getIndexFileInput(indexID, snapshot, file);
		byte[] buffer = new byte[16384];
		OutputStream outputStream = response.getOutputStream();
		long length = indexInput.length();
    long readCount = 0;
    while (readCount < length) {
      int toRead = readCount + 16384 > length ? (int)(length - readCount) : 16384;
      indexInput.readBytes(buffer, 0, toRead);
      outputStream.write(buffer, 0, toRead);
      readCount += toRead;
    }
	}
	
	private void handleGetShardInfos(HttpServletRequest request, PrintStringWriter printStringWriter) throws Exception {
		Map<String,ShardInfo> map = searchService.getShardInfos();
		OceanXML oceanXML = new OceanXML();
		oceanXML.addHeader(ClusterConstants.CELL_TRANSACTION_ID_FIELD, searchService.getHighestCellTransactionIDDatabase());
		oceanXML.setServerInfo(searchService.getServerInfo());
		oceanXML.addBodyAll(map.values());
		oceanXML.write(printStringWriter);
	}
	
	private void handleGetTransactionRecords(HttpServletRequest request, PrintStringWriter printStringWriter) throws Exception {
		String shardid = request.getParameter("shardid");
		CLong length = ClusterUtil.getLong("length", request);
		TransactionID startid = ClusterUtil.getTransactionID("startid", request);
		List<TransactionRecord> transactionRecords = searchService.getTransactionRecords(shardid, startid, length);
		OceanXML.write(searchService.getServerInfo(), transactionRecords, printStringWriter);
	}
	
	private void handleGetTermFrequencies(HttpServletRequest request, PrintStringWriter writer) throws Exception {
		String q = request.getParameter("q");
		Query query = ClusterUtil.parseQuery(q, searchService.getIndexSchema());
		TimeoutState timeoutState = new TimeoutState(1000 * 5);
		TermFrequencies termFrequencies = searchService.getTermFrequencies(query, timeoutState);
		List<Element> elements = new ArrayList<Element>();
		for (Map.Entry<Term, AtomicLong> entry : termFrequencies.entrySet()) {
			Term term = entry.getKey();
			AtomicLong number = entry.getValue();
			Element element = new Element("termfrequency");
			XMLUtil.setAttribute("field", term.field(), element);
			XMLUtil.setAttribute("text", term.text(), element);
			XMLUtil.setAttribute("frequency", number, element);
			elements.add(element);
		}
		OceanXML.write(searchService.getServerInfo(), elements, writer);
	}

	public void destroy() {
		try {
			searchService.stop();
		} catch (Throwable throwable) {
			log.log(Level.SEVERE, "", throwable);
		}
		super.destroy();
	}

	public void init() throws ServletException {
		try {
			searchService = SearchService.getSearchService();
			log.info("SearchServlet.init finished");
		} catch (Throwable ex) {
			log.log(Level.SEVERE, "", ex);
		}
	}
}
